package baiduschPro;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import com.jayway.jsonpath.JsonPath;

import hylanda.common.Transfer2JSON;
import net.minidev.json.JSONArray;
import net.minidev.json.JSONObject;
import net.minidev.json.JSONValue;

@Tags({"data format exchagent "})
@CapabilityDescription("data format exchagent")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})

public class ExchangeDataProcessor extends AbstractProcessor {

	  
	 public static final PropertyDescriptor ISGETPART = new PropertyDescriptor
	            .Builder().name("ISGETPART")
	            .displayName("Is get data part")
	            .description("Is get data part by jsonpath")
	            .required(true)
	            .defaultValue("false")
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor PARTSCHEME = new PropertyDescriptor
	            .Builder().name("PARTSCHEME")
	            .displayName("partscheme")
	            .description("如何获取部分数据，用jsonpath格式表示")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 public static final PropertyDescriptor MYSCHEME = new PropertyDescriptor
	            .Builder().name("MYSCHEME")
	            .displayName("My scheme")
	            .description("每条数据中字段转化，例如coname=schword")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYSCHEMEADD = new PropertyDescriptor
	            .Builder().name("MYSCHEMEADD")
	            .displayName("My scheme add")
	            .description("每条数据中新增的字段和值,例如：source_type=1,rel_type=\"m\",加引号表示字符串，不加表示数字")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor ISSPLIT = new PropertyDescriptor
	            .Builder().name("ISSPLIT")
	            .displayName("is split")
	            .description("多条记录是否拆分成单条")
	            .required(false)
	            .defaultValue("0")
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor ADDHIGHTLAYER = new PropertyDescriptor
	            .Builder().name("ADDHIGHTLAYER")
	            .displayName("addHightlayer")
	            .description("上面加上一层")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 public static final PropertyDescriptor ADDHIGHTLAYEREX = new PropertyDescriptor
	            .Builder().name("ADDHIGHTLAYEREX")
	            .displayName("addHightlayerEx")
	            .description("上面加上一层的下一层为object还是list,0 保持原样 ，1转成list")
	            .defaultValue("1")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 //
	 
	public static final Relationship MY_RELATIONSHIPNO = new Relationship.Builder()
	            .name("gen failure")
	            .description("failure relationship")
	            .build();

	public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
            .name("gen sucess")
            .description("sucess relationship")
            .build();

	private List<PropertyDescriptor> descriptors;

	private Set<Relationship> relationships;
	    
	public ExchangeDataProcessor() {
		// TODO Auto-generated constructor stub
	}
	
	
	 @Override
	    protected void init(final ProcessorInitializationContext context) {
	        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	        descriptors.add(MYSCHEME);
	        descriptors.add(MYSCHEMEADD);
	        descriptors.add(ISGETPART);
	        descriptors.add(PARTSCHEME);
	        descriptors.add(ISSPLIT);
	        descriptors.add(ADDHIGHTLAYER);
	        descriptors.add(ADDHIGHTLAYEREX);
	        
	       
	        this.descriptors = Collections.unmodifiableList(descriptors);

	        final Set<Relationship> relationships = new HashSet<Relationship>();
	        relationships.add(MY_RELATIONSHIPOK);
	        relationships.add(MY_RELATIONSHIPNO);
	        this.relationships = Collections.unmodifiableSet(relationships);  
	    }

	    @Override
	    public Set<Relationship> getRelationships() {
	        return this.relationships;
	    }

	    @Override
	    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
	        return descriptors;
	    }

	    @OnScheduled
	    public void onScheduled(final ProcessContext context) {

	  //  	System.out.println("onScheduled is clled");
	    }

	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
		//a=b,c=d,f=e
		FlowFile flowFile = session.get();
        if ( flowFile == null ) {
        	
        	System.out.println("flowfile is null");
           return;
        }
        //读取FlowFile
        String content = "";
        InputStream in = session.read(flowFile);
        try {
			byte[] by = readStream(in);
			content = new String(by,"UTF-8");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        
		 PropertyValue myp = context.getProperty("MYSCHEME");
		 PropertyValue mypartp = context.getProperty("ISGETPART");
		 PropertyValue partp = context.getProperty("PARTSCHEME");
		 PropertyValue issplitp = context.getProperty("ISSPLIT");
		 PropertyValue addhlp = context.getProperty("ADDHIGHTLAYER");
		 PropertyValue myaddp = context.getProperty("MYSCHEMEADD");
		 PropertyValue addhlexp = context.getProperty("ADDHIGHTLAYEREX");
		 
		 String isSplit = issplitp.getValue();
		 if(isSplit == null)
			 isSplit = "0";
		 
	     String myscheme = myp.getValue();
	     if(myscheme == null)
	    	 myscheme = "";
	     String ispart = mypartp.getValue();
	     if(ispart == null)
	    	 ispart = "";
	     
	     String partscheme = partp.getValue();
	     if(partscheme == null)
	    	 partscheme = "";
	     
	     String hiname = addhlp.getValue();//上层加上，最后处理
	     if(hiname == null)
	    	 hiname = "";
	     
	     String addstr = myaddp.getValue();
	     if(addstr == null)
	    	 addstr = "";
	     
	     String addex = addhlexp.getValue();
	     
	     String jsonresult = content;
	     if(ispart.equals("true"))
	     {
	    	 jsonresult = exGetDataPartJsonPath(partscheme,content);
	    	 
	    	 if(!myscheme.isEmpty() || !addstr.isEmpty())
	    	 {
	    		 jsonresult = exChangeData(myscheme,addstr,jsonresult);
	    	 }
	     }
	     else
	     {
	    	 jsonresult = exChangeData(myscheme,addstr,content);
	    	 
	     }
	     if(!hiname.isEmpty())
	     {
	    	 if(!isSplit.equalsIgnoreCase("1"))//先分割，再加高一层
	    		 jsonresult = addHightLayer(jsonresult,hiname,addex); 
	     }
	     
	    
	     final String strojcontent = jsonresult;

	     int nlen = strojcontent.length();
	     if(nlen <= 2)
	     {
	    	//写flowfile
		     flowFile= session.write(flowFile, new OutputStreamCallback(){
		    		@Override
		    		public void process(OutputStream out) throws IOException{
		    			out.write(strojcontent.getBytes("UTF-8"));
		    		}
		    	});
		     
		     session.transfer(flowFile,MY_RELATIONSHIPNO);
	     }
	     else
	     {
	    	 if(isSplit.equalsIgnoreCase("1"))
	    	 {
	    		 Object objtmp = JSONValue.parse(strojcontent) ;
	    		 String classname = objtmp.getClass().getName();
	    		 if(classname.equalsIgnoreCase("net.minidev.json.JSONArray"))
	    		 {
	    			 
	    			 List list = (List)objtmp;
	    			 for(int i = 0 ; i < list.size();i++)
	    			 {
	    				 JSONObject objone = (JSONObject) list.get(i);
	    				 String onetmp = objone.toJSONString();
	    				 if(!hiname.isEmpty())
	    					 onetmp = addHightLayer(onetmp,hiname,addex);
	    				 final String  one = onetmp;
	    				 if(i> 0)
	    				 {
	    					 flowFile = session.create();
	    				 }
	    				 
	    				//写flowfile
					     flowFile= session.write(flowFile, new OutputStreamCallback(){
					    		@Override
					    		public void process(OutputStream out) throws IOException{
					    			out.write(one.getBytes("UTF-8"));
					    		}
					    	});
					     
					     session.transfer(flowFile,MY_RELATIONSHIPOK);
	    			 }//for i
	    			 
	    		 }
	    		 else
	    		 {
	    			//写flowfile
				     flowFile= session.write(flowFile, new OutputStreamCallback(){
				    		@Override
				    		public void process(OutputStream out) throws IOException{
				    			out.write(strojcontent.getBytes("UTF-8"));
				    		}
				    	});
				     
				     session.transfer(flowFile,MY_RELATIONSHIPOK);
	    		 }

	    	 }
	    	 else
	    	 {
			     //写flowfile
			     flowFile= session.write(flowFile, new OutputStreamCallback(){
			    		@Override
			    		public void process(OutputStream out) throws IOException{
			    			out.write(strojcontent.getBytes("UTF-8"));
			    		}
			    	});
			     
			     session.transfer(flowFile,MY_RELATIONSHIPOK);
	    	 }
	     }
	}
	
	public static byte[] readStream(InputStream inStream) throws Exception {
        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
        byte[] buffer = new byte[1024];  
        int len = -1;  
        while ((len = inStream.read(buffer)) != -1) {  
            outSteam.write(buffer, 0, len);  
        }  
        outSteam.close();  
        inStream.close();  
        return outSteam.toByteArray();  
    }  
	
	//单条数据
	public String exChangeData(String scheme,String addScheme,String content)
	{
		String contentok = "";
		String[] split = scheme.split(",");
		String[] addsplit = addScheme.split(",");
	     
		Object  obj= JSONValue.parse(content);
		String classname = obj.getClass().getName();
		if(classname.equals("net.minidev.json.JSONObject"))
		{
			JSONObject objnew = exChangeOneData((JSONObject)obj,split,addsplit);
			contentok = objnew.toJSONString();
		}
		else if(classname.equals("net.minidev.json.JSONArray"))
		{
			JSONArray arr = new JSONArray();
			List list = (List)obj;
			for(int i = 0; i < list.size(); i++)
			{
				JSONObject one = (JSONObject) list.get(i);
				JSONObject objnew = exChangeOneData(one,split,addsplit);
				arr.add(objnew);
			}//for i
			contentok = arr.toJSONString();
		}
		else
		{}
	     return contentok;
	}
	
	private JSONObject exChangeOneData(JSONObject obj,String[] split,String[] addsplit)
	{
		JSONObject objnew = new JSONObject();
		if(obj == null)
			return null;
		
	     for(int i = 0; i < split.length; i++)
	     {
	    	String one = split[i];
	    	String splitone[] = one.split("=");
	    	if(splitone.length == 2)
	    	{
	    		String before = splitone[0];
	    		String after = splitone[1];
	    		Object objone=obj.get(before);
	    		objnew.put(after, objone);
	    	}
	    	else
	    	{}
	     }//for i
	     
	     for(int j = 0; j < addsplit.length; j++)
 		{
 			String onetmp = addsplit[j];
 			String[] onetmpsplit = onetmp.split("=");
 			if(onetmpsplit.length == 2)
 			{
 				String key = onetmpsplit[0];
 				String value = onetmpsplit[1];
 				Object objv = null;
 				int nlen = value.length();
 				
 				int nvalue = 0;
 				char first,last;
 				if(nlen > 1)
 				{
 					first=value.charAt(0);
 					last = 	value.charAt(nlen-1);
 					if(first =='\"' && last == '\"')
 					{
 						value = value.substring(1, nlen-1);
 						objv = value;
 					}
 					else
 					{
 						nvalue = Integer.parseInt(value);
 						objv = nvalue;
 					}
 						
 				}
 				else if(nlen == 1)
 				{
 					nvalue = Integer.parseInt(value);
						objv = nvalue;
 				}
 				else
 					continue;
 				objnew.put(key, objv);
 			}
 		}//for j
	    
	     return objnew;
	}
	
	//获得部分存储
	private String exGetDataPartJsonPath(String myscheme,String json)
	{
		String str = "";		
	    Object obj = JsonPath.read(json, myscheme);
	    
	    if(obj != null)
	    {
	    	String classname = obj.getClass().getName();//java.util.LinkedHashMap   //net.minidev.json.JSONArray
	    	if(classname.equals("java.util.LinkedHashMap") || classname.equals("java.util.HashMap") || classname.equals("java.util.Map"))
	    	{
	    		str = Transfer2JSON.TransferMap2JSON((java.util.Map)obj);
	    	}
	    	else if(classname.equals("java.util.List") || classname.equals("java.util.ArrayList") || classname.equals("java.util.LinkedList"))
	    	{
	    		str =  Transfer2JSON.TransferMap2JSONListEx((java.util.List)obj);
	    	}
	    	else
	    		str = obj.toString();
//	    	System.out.println(classname);
	    	//	    str =  Transfer2JSON.TransferMap2JSON();
	    }
	    	
		 return str;
	}
	
	//addtype  上面加上一层的下一层为object还是list,0 保持原样 ，1转成list
	private String addHightLayer(String jsondata,String hightname,String addtype)
	{
		String strResult = "";
		Object  obj= JSONValue.parse(jsondata);
		JSONObject objh = new  JSONObject();
		if(addtype.equalsIgnoreCase("0"))
			objh.put(hightname, obj);
		else if(addtype.equalsIgnoreCase("1"))//list
		{
			JSONArray list = new JSONArray();
			list.add(obj);
			objh.put(hightname, list);
		}
		else
			objh.put(hightname, obj);
		
		strResult = objh.toJSONString();
		return strResult;
	}

}
